package com.atguigu.bigdata.spark.core.acc;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.sources.In;
import org.apache.spark.util.LongAccumulator;
import scala.Tuple2;

import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.function.Consumer;

public class Spark05_Bc_JAVA {
    public static void main(String[] args) {
        // 1.创建配置对象
        SparkConf conf = new SparkConf().setMaster("local[*]").setAppName("sparkCore");

        // 2. 创建sparkContext
        JavaSparkContext sc = new JavaSparkContext(conf);

        List<Tuple2<String, Integer>>  list = Arrays.asList(new Tuple2<String, Integer>("a", 1),
                new Tuple2<String, Integer>("b", 2), new Tuple2<String, Integer>("c", 3));
        JavaPairRDD<String, Integer> rdd1 = sc.parallelizePairs(list);
        Map<String , Integer> map = new HashMap<>();
        map.put("a", 5);
        map.put("b",6);
        map.put("c",7);

        Broadcast<Map<String, Integer>> bc = sc.broadcast(map);
        JavaPairRDD<String, Tuple2<Integer,Integer>> rdd2 = rdd1.mapToPair(new PairFunction<Tuple2<String, Integer>, String, Tuple2<Integer, Integer>>() {
            @Override
            public Tuple2<String, Tuple2<Integer, Integer>> call(Tuple2<String, Integer> inputTuple) throws Exception {
                String key = inputTuple._1;
                int val = inputTuple._2;

                if(bc.value().containsKey(key)) {
                    Tuple2<Integer, Integer> t1 = new Tuple2<>(val, bc.value().get(key));
                    Tuple2<String, Tuple2<Integer, Integer>> res = new Tuple2<>(key, t1);
                    return res;
                }
                return null;
            }
        });

        rdd2.collect().forEach(new Consumer<Tuple2<String, Tuple2<Integer, Integer>>>() {
            @Override
            public void accept(Tuple2<String, Tuple2<Integer, Integer>> stringTuple2Tuple2) {
                System.out.println(stringTuple2Tuple2.toString());
            }
        });
        sc.stop();

    }
}
